package com.hex.es.bulk;

import com.alibaba.fastjson.JSON;
import com.hex.es.component.EsBatchRouteComponent;
import com.hex.es.entity.EsInfoDomain;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * es信息逻辑处理类
 * 封装一层对es操作的处理
 *
 * @author hex
 * @date 2019/11/01
 */
@Slf4j
@Component
public class EsOperationLogicRunnable {

    /**是否允许执行，保证只有一个线程执行*/
    private static AtomicBoolean isRun = new AtomicBoolean(Boolean.TRUE);

    @Resource
    private EsBatchRouteComponent esBatchRouteComponent;

    public void run() {
        List<EsInfoDomain> esInfoDomains = null;
        try {
            if (!isRun.get() || !isRun.compareAndSet(Boolean.TRUE, Boolean.FALSE)) {
                //被其他线程抢占，所以当前线程直接返回释放
                return;
            }
            esInfoDomains = EsInfoStorageCache.get(EsOperationControlCenter.getThreshold());
            log.info("EsOperationLogicRunnable get data,start doBatch。data:{}", JSON.toJSONString(esInfoDomains));
            doBatch(esInfoDomains);
        } catch (Exception e) {
            EsInfoStorageCache.addAllOfErrorQueue(esInfoDomains);
            log.error("es信息逻辑处理异常",e);
        } finally {
            isRun.compareAndSet(Boolean.FALSE,Boolean.TRUE);
        }
    }

    /**
     * 进行批量处理
     * 1、把传过来的一组批量数据按操作类型分开，然后一批批去处理
     * @param esInfoDomains
     */
    private void doBatch(List<EsInfoDomain> esInfoDomains) {
        List<EsInfoDomain> operationList = new ArrayList<>();
        for (int i=0;i<esInfoDomains.size();i++) {
            EsInfoDomain esInfoDomain = esInfoDomains.get(i);
            operationList.add(esInfoDomain);
            if (i+1 == esInfoDomains.size()) {
                //已经遍历该list到达最后一个元素
                try {
                    //处理
                    esBatchRouteComponent.route(operationList);
                } catch (Exception e) {
                    log.error("EsOperationLogicRunnable doBatch error!",e);
                }
                return;
            }
            EsInfoDomain esInfoDomain1 = esInfoDomains.get(i + 1);
            if (!esInfoDomain.getOperationEnum().equals(esInfoDomain1.getOperationEnum())) {
                try {
                    //处理
                    esBatchRouteComponent.route(operationList);
                } catch (Exception e) {
                    log.error("EsOperationLogicRunnable doBatch error!",e);
                } finally {
                    //清空list
                    operationList.clear();
                }
            }
        }
    }
}
